package com.smsc.headend.task.engine.service.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smsc.headend.common.constant.CacheNameConstant;
import com.smsc.headend.module.data.dto.MeterProgressDTO;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.meterscheme.entity.CollMeterTaskPlan;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.task.engine.mapper.CollMeterTaskPlanMapper;
import com.smsc.headend.task.engine.service.CollMeterTaskPlanService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.CacheConfig;
import org.springframework.cache.annotation.CachePut;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author liangli
 * @date 2020/9/2
 */
@Service
@Slf4j
@CacheConfig(cacheNames = CacheNameConstant.UTE_METER_PLAN)
public class CollMeterTaskPlanServiceImpl extends ServiceImpl<CollMeterTaskPlanMapper, CollMeterTaskPlan> implements CollMeterTaskPlanService {

    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    DataSource dataSource;

    @Autowired
    @Lazy
    CollMeterTaskPlanService self;


    @Override
    public void saveMeterCollectProgress(Long tv, Long meterId, Long lpDataItemId) {
        MeterProgressDTO meterProgressDTO = new MeterProgressDTO();
        meterProgressDTO.setLpDataItemId(lpDataItemId);
        meterProgressDTO.setMeterId(meterId);
        meterProgressDTO.setProgressTv(tv);
        kafkaTemplate.send(TaskKafkaTopic.COLLECT_PROGRESS_PLAN_UPDATE, Convert.toStr(meterId), JSONUtil.toJsonStr(meterProgressDTO).getBytes(Charset.forName("utf-8")));
    }


    /**
     * meter 和负荷曲线要求为一组
     *
     * @param meterReadsLps
     */
    @Override
    @Transactional
    public void dataProgressIntegrityUpdate(List<RdMeterReadsLp> meterReadsLps) {
        if (CollectionUtils.isEmpty(meterReadsLps)) {
            return;
        }
        /**
         * key: meterId + ":" + lpDataItemId
         * value: Set 【tvs】
         */
        Map<String, Set<Long>> meterLpDataTvMap = new HashMap<>();
        for (RdMeterReadsLp meterReadsLp : meterReadsLps) {
            String meterIdLpDataItemId = meterReadsLp.getMeterId() + ":" + meterReadsLp.getLpDataItemId();
            if (!meterLpDataTvMap.containsKey(meterIdLpDataItemId)) {
                meterLpDataTvMap.put(meterIdLpDataItemId, new HashSet<Long>());
            }
            meterLpDataTvMap.get(meterIdLpDataItemId).add(meterReadsLp.getTv());
        }
        kafkaTemplate.send(TaskKafkaTopic.LP_DATA_INTEGRITY_CALCULATE, JSONUtil.toJsonStr(meterLpDataTvMap).getBytes(StandardCharsets.UTF_8));
    }

    @Override
    @Cacheable(key = "#meterId + '.' + #profileDataItemId", unless = "#result == null")
    public CollMeterTaskPlan getCollMeterTaskPlan(Long meterId, Long profileDataItemId) {
        return lambdaQuery().eq(CollMeterTaskPlan::getMeasurementProfileId, profileDataItemId).eq(CollMeterTaskPlan::getMeterId, meterId).one();
    }

    @Override
    @CachePut(key = "#collMeterTaskPlan.meterId + '.' + #collMeterTaskPlan.measurementProfileId", unless = "#result == null")
    public CollMeterTaskPlan updateCache(CollMeterTaskPlan collMeterTaskPlan) {
        return collMeterTaskPlan;
    }

    @Override
    @Transactional
    public Boolean batchSaveMeterCollectProgress(List<MeterProgressDTO> dtoList) {
        List<CollMeterTaskPlan> updateList = new ArrayList<>();
        List<CollMeterTaskPlan> insertList = new ArrayList<>();
        Long startTime = SystemClock.now();
        Long now = startTime / DateUtils.MILLIS_PER_SECOND;
        for (MeterProgressDTO meterProgressDTO : dtoList) {
            CollMeterTaskPlan collMeterTaskPlan = self.getCollMeterTaskPlan(meterProgressDTO.getMeterId(), meterProgressDTO.getLpDataItemId());
            if (collMeterTaskPlan != null) {
                collMeterTaskPlan.setUpdateTime(now);
                collMeterTaskPlan.setProgressTime(meterProgressDTO.getProgressTv());
                updateList.add(collMeterTaskPlan);
            } else {
                CollMeterTaskPlan collMTP = new CollMeterTaskPlan();
                collMTP.setMeterId(meterProgressDTO.getMeterId());
                collMTP.setMeasurementProfileId(meterProgressDTO.getLpDataItemId());
                collMTP.setProgressTime(meterProgressDTO.getProgressTv());
                collMTP.setUpdateTime(now);
                insertList.add(collMTP);
            }
        }
        Long queryEnd = SystemClock.now();
        if (CollectionUtils.isNotEmpty(insertList)) {
            Map<String, List<CollMeterTaskPlan>> collect = insertList.stream().collect(Collectors.groupingBy(s -> s.getMeterId() + ":" + s.getMeasurementProfileId()));
            List<CollMeterTaskPlan> distinctMaxPlanList = collect.entrySet().stream().map(e -> e.getValue().stream().reduce((maxPlan, currentPlan) -> {
                if (currentPlan.getProgressTime() > maxPlan.getProgressTime()) {
                    return currentPlan;
                }
                return maxPlan;
            }).get()).collect(Collectors.toList());
            this.executeBatch(distinctMaxPlanList,  (session, collMeterTaskPlan) -> {
                session.insert("com.smsc.headend.task.engine.mapper.CollMeterTaskPlanMapper.insert", collMeterTaskPlan);
            });
        }
        if (CollectionUtils.isNotEmpty(updateList)) {
            Map<String, List<CollMeterTaskPlan>> collect = updateList.stream().collect(Collectors.groupingBy(s -> s.getMeterId() + ":" + s.getMeasurementProfileId()));
            List<CollMeterTaskPlan> distinctMaxPlanList = collect.entrySet().stream().map(e -> e.getValue().stream().reduce((maxPlan, currentPlan) -> {
                if (currentPlan.getProgressTime() > maxPlan.getProgressTime()) {
                    return currentPlan;
                }
                return maxPlan;
            }).get()).collect(Collectors.toList());

            this.executeBatch(distinctMaxPlanList,  (session, collMeterTaskPlan) -> {
                session.insert("com.smsc.headend.task.engine.mapper.CollMeterTaskPlanMapper.updateProgressTime", collMeterTaskPlan);
            });
        }
        Long cacheStart = SystemClock.now();
        updateList.forEach(self::updateCache);
        log.info("updateProgressTime: insertSize:{},updateSize:{},cost:total={}ms,query：{}ms,execute:{}ms,cache:{}ms", insertList.size(), updateList.size()
                , SystemClock.now() - startTime, queryEnd - startTime, cacheStart - queryEnd, SystemClock.now() - cacheStart);
        return true;
    }

}
